package com.itmentu.controller;

import org.apache.kafka.common.utils.Utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @Author：http://www.itmentu.com
 * @date： 2021/12/18 16:45
 * @Description：
 ***/
@RestController
@RequestMapping(value = "/producer")
public class SendController {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /***
     * 发送消息
     * topic:要发送的队列
     * msg:发送的消息
     */
    @GetMapping(value = "/send/{topic}/{msg}")
    public String send(@PathVariable(value = "topic")String topic,@PathVariable(value = "msg")String msg) throws Exception{
        //消息发送
        kafkaTemplate.send(topic, "a",msg);
        return "SUCCESS";
    }

    /***
     * 发送消息
     * topic:要发送的队列
     * msg:发送的消息
     */
    @GetMapping(value = "/async/send/{topic}/{msg}")
    public String asyncsend(@PathVariable(value = "topic")String topic,@PathVariable(value = "msg")String msg) throws Exception{
        //打印时间，方便查看是否是异步
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss.SSSZ");
        //消息发送，获取返回对象Future
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, msg);
        //添加异步获取结果方法
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println(simpleDateFormat.format(new Date())+"--->主题："+topic+"消息发送失败，发送消息为："+msg);
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println(simpleDateFormat.format(new Date())+"--->主题："+topic+"消息发送成功，发送消息为："+msg+"，偏移量:"+result.getRecordMetadata().offset());
            }
        });
        System.out.println(simpleDateFormat.format(new Date())+"--------发送消息完成-----");
        return "SUCCESS";
    }


    /***
     * 开启事务发送消息
     * topic:要发送的队列
     * msg:发送的消息
     */
    @GetMapping(value = "/transaction/send/{topic}/{msg}")
    public String transactionSend(@PathVariable(value = "topic")String topic,@PathVariable(value = "msg")String msg) throws Exception{
        //事务-多消息发送
        kafkaTemplate.executeInTransaction(operations ->{
                    operations.send(topic,"hello111!"+msg);
                    operations.send(topic,"hello122!"+msg);
                    operations.send(topic,"hello133!"+msg);
                    if(msg.equals("java")){
                        throw new RuntimeException("回滚，所有消息都不会发送");//模拟报错
                    }else{
                        return true;
                    }
                });
        return "SUCCESS";
    }

    /***
     * 开启事务发送消息
     * topic:要发送的队列
     * msg:发送的消息
     */
    @Transactional
    @GetMapping(value = "/annotation/transaction/send/{topic}/{msg}")
    public String annotationTransactionSend(@PathVariable(value = "topic")String topic,@PathVariable(value = "msg")String msg) throws Exception{
        //发送消息
        kafkaTemplate.send(topic,msg);

        //模拟报错，测试事务
        if(msg.equals("java")){
            throw new RuntimeException("回滚，所有消息都不会发送");
        }
        return "SUCCESS";
    }
}
